package topicmanager;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.config.TopicConfig;

import java.util.Arrays;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;

/**
 * @Author:RenPu
 * @Date: 2020/3/18 19:50
 * @Version: 1.0
 * @description:topic主题操作，创建，删除
 */
public class TopicManagerDemo {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        
        //创建配置对象
        Properties properties = new Properties();

        //配置Kafka集群参数
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092");
        
        //创建AdminClient对象
        AdminClient client = AdminClient.create(properties);

        //创建主题
        client.createTopics(Arrays.asList(new NewTopic("t2",3,(short)3)));

        //展示所有的主题方法
        ListTopicsResult result = client.listTopics();
        Set<String> topics = result.names().get();

        //Lamda表达式
        //topics.forEach(topic-> System.out.println(topic));

        for (String topic : topics) {
            System.out.println(topic);
        }

        //查看topic详情
        DescribeTopicsResult result1 = client.describeTopics(Arrays.asList("t1"));
        Map<String, KafkaFuture<TopicDescription>> values = result1.values();
        values.forEach((k,v)-> System.out.println(k+"|"+v));
        for (Map.Entry<String,KafkaFuture<TopicDescription>> entry: values.entrySet()) {
            System.out.println(entry.getKey()+"|"+entry.getValue().get());
        }
        
        
        //删除指定的topic
        client.deleteTopics(Arrays.asList("t2"));


        //释放资源
        client.close();


    }
}
